package com.practice.car.cardataapp.schelduer.spark.task.sale;

import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.SaleRunner;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.*;
import org.apache.spark.sql.*;

import org.apache.spark.sql.catalyst.encoders.RowEncoder;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Serializable;
import scala.Tuple2;

import java.math.BigDecimal;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

import static com.practice.car.cardataapp.schelduer.spark.Schedule.getConnection;

public class App implements Serializable, SaleRunner {
    private Connection conn = null;

    @Override
    public void run() {
        SparkSession sparkSession = SparkSession
                .builder()
                .master(Schedule.sparkMaster)
                .appName("SparkAnalyzer")
                .getOrCreate();
        String path = Schedule.saleHdfsUrl;
        //文件位置需要修改！！！
        Dataset<Row> ds = sparkSession.read().json(path);
        //统计总销量，此处的总销量是只给getBrandSales用的
        Integer totalSale = getTotalSale(ds);
        //统计各车型销量数据
        getBrandSales(ds, totalSale);
        //getCarTotalSaleDataMonthly(ds,totalSale);
        getCarTotalSaleDataMonthly2(ds);
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        sparkSession.stop();
    }

    //统计总销量，给getBrandSales用的
    public Integer getTotalSale(Dataset<Row> ds) {
        Integer totalSale = ds.select("sale").map((MapFunction<Row, Integer>) row -> {
            if (row.getAs("sale") == null) {
                return 0;
            }
            String saleString = row.getAs("sale").toString();
            return Integer.valueOf(saleString);
        }, Encoders.INT()).reduce((ReduceFunction<Integer>) (t1, t2) -> t1 + t2);
        return totalSale;
    }
    public void getCarTotalSaleDataMonthly2(Dataset<Row> ds) {

        /*
        为ds2构建列表
         */
        List<StructField> listOfStructField = new ArrayList<StructField>();
        listOfStructField.add(DataTypes.createStructField("date", DataTypes.StringType, true));
        listOfStructField.add(DataTypes.createStructField("year", DataTypes.StringType, true));
        listOfStructField.add(DataTypes.createStructField("sale", DataTypes.IntegerType, true));
        StructType structType = DataTypes.createStructType(listOfStructField);


        //107行 withColumn新增列"year",而"year"是从"date"拷贝而来（包括数据，所以此时二者只是列名不同），filter筛选掉year和date里面为null的数据
        //ds2是从ds选择date，sale列，新增year列，然后去除null数据，然后构建成dateset<Row>
        Dataset<Row> ds2 = ds.select("date", "sale").withColumn("year", ds.col("date")).filter("year is not null and date is not null").map((MapFunction<Row, Row>) row -> {
            String dateString = row.getAs("date");
            String yearString = row.getAs("year").toString().split("-")[0];//从20xx-0x-xx取出年20xx
            String saleString = row.getAs("sale").toString();
            return RowFactory.create(dateString, yearString, Integer.valueOf(saleString));
        }, RowEncoder.apply(structType));


        Dataset<Row> yearSumSaleData = ds2.groupBy("year").sum("sale");//按年统计销量，此处是sale是年销量
        Dataset<Row> dateSumSaleData = ds2.groupBy("date").sum("sale");//按具体日期统计销量，此处的sale是月销量


        Dataset<Row> dateTotalSaleData = dateSumSaleData.withColumn("year", dateSumSaleData.col("date")).map((MapFunction<Row, Row>) row -> {
            String dateString = row.getAs("date");
            String yearString = row.getAs("year").toString().split("-")[0];//从20xx-0x-xx取出年20xx
            String saleString = row.getAs("sum(sale)").toString();
            return RowFactory.create(dateString, yearString, Integer.valueOf(saleString));
        }, RowEncoder.apply(structType));

        //类似数据库的join操作
        //（year sale【年销量】）join（year date sale【月销量】）--->
        dateTotalSaleData.join(yearSumSaleData, dateTotalSaleData.col("year").equalTo(yearSumSaleData.col("year")), "inner")
                .foreach((ForeachFunction<Row>) row -> {
                    String month = row.getAs("date").toString();    //month:【20xx-xx】
                    Integer sale = Integer.valueOf(row.getAs("sale").toString());   //sale：月销量
                    Integer wholeYearSale = Integer.valueOf(row.getAs("sum(sale)").toString()); //wholeYearSale：月份所在那年的年销量
                    double proportion = BigDecimal.valueOf(((double) sale / wholeYearSale) * 100).setScale(1, BigDecimal.ROUND_HALF_UP).doubleValue();
                    String sql = "insert into car_sale_month (month , sale,proportion) values(" +
                            "\"" + month + "\",\"" + sale + "\",\"" + proportion + "\")";
                    Connection conn = getConnection();
                    conn.createStatement().execute(sql);
                }
        );
    }

    //全国各个品牌的销量和各品牌销量占总销量的比例
    public void getBrandSales(Dataset<Row> ds, Integer totalSale) {

        JavaPairRDD saleCount = ds.select("brand", "sale").javaRDD().mapToPair((PairFunction<Row, String, Integer>) row -> {
            String brandString = row.getAs("brand");
            String saleString = null;
            if (row.getAs("sale") == null) {
                saleString = "0";
            } else {
                saleString = row.getAs("sale").toString();
            }
            return new Tuple2<String, Integer>(brandString, Integer.valueOf(saleString));
        }).reduceByKey((Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2);
        //统计比例；写入数据库
        saleCount.foreach((VoidFunction<Tuple2<String, Integer>>) t -> {
            //proportion  品牌的销量占总销量(totalSale)的百分比
            //t._1  brand   String
            //t._2  sale    Integer
            double proportion = BigDecimal.valueOf((t._2.doubleValue() / totalSale) * 100)
                    .setScale(1, BigDecimal.ROUND_HALF_UP).doubleValue();
            //占比大于0的数据才入库
            if (proportion > 0.0) {
                String sql = "insert into brand_sales (brand, sale,proportion) values(" +
                        "\"" + t._1 + "\",\"" + t._2 + "\",\"" + proportion + "\")";
                Connection conn = getConnection();
                conn.createStatement().execute(sql);
            }
        });
    }
}
